#!/bin/bash
# File Name: diff_count.sh
# Author: moshan
# mail: mo_shan@yeah.net
# Created Time: 2019-02-22 16:46:26
# Function: 对比master与slave的数据行是否一致
#########################################################################
work_dir="/data/git/check_data_for_mysql"
. ${work_dir}/conf/check.conf
db="${1}"
this_thread_num="${2}"
thread_file="${3}"
mysql_ver="${4}"
tmp_file="${pri_file_dir}/pri_${db}.log"
pid_file="${pid_dir}/${db}.pid"
table_list_file="${list_dir}/${db}.list"
[ ! -f "${table_list_file}" ] && touch ${table_list_file}
function f_check_diff()
{
	db_name="${1}"
	mysql_comm="${mysql_path} -u${mysql_user} -p${mysql_passwd} -h${host_list[0]} -P${mysql_port1} ${db_name} -A"
	table=($(${mysql_comm} -NBe "show tables" 2>/dev/null))
	for ((i=0;i<${#table[@]};i++))
	do
		table_check_stat="$(grep -c "${db_name}\.${table[${i}]}" ${table_list_file})"
		[ "${table_check_stat}x" != "0x" ] && continue
		mysql_comm="${mysql_path} -u${mysql_user} -p${mysql_passwd} -h${host_list[0]} -P${mysql_port1} ${db_name} -A"
		pri_name="$(${mysql_comm} -NBe "desc ${table[${i}]}" 2>/dev/null|awk 'BEGIN{str2="PRI";pri=0;}{str=$4;if (str==str2 && pri==0){pri+=1;pri_str=(pri_str"+"$1);pri_type=(pri_type"+"$2)}}END{if(pri>0){print pri_str":"pri_type":"pri}else{print "NULL"}}')"
		pri_type=($(awk -F: '{print $2}' <<< "${pri_name}"|tr "+" " "))
		pri_count="$(awk -F: '{print $3}' <<< "${pri_name}")"
		pri_name=($(awk -F: '{print $1}' <<< "${pri_name}"|tr "+" " "))
		select_str=""
		[ "${pri_name}x" == "NULLx" ] && pri_count=0
		for ((tmp=0;tmp<${pri_count};tmp++))
		do
			if [ "$(grep -c "int" <<< "${pri_type[${tmp}]}")x" != "1x" ]
			then
				select_str="${select_str},concat(\"'\",${pri_name[${tmp}]},\"'\")"
			else
				select_str="${select_str},${pri_name[${tmp}]}"
			fi
		done
		select_str="$(sed 's/^,//g' <<< "${select_str}")"
		if [ "${pri_name}x" != "NULLx" -o "${check_no_pk}x" == "1x" ]
		then
			pri_name="($(echo "${pri_name[@]}"|tr " " ","))"
			> ${tmp_file}
			tmp_file_count=0
			tmp_file_count_tmp=0
			if [ "${pri_count}x" != "0x" ]
			then
				sql_str_tmp="select ${select_str} from ${table[${i}]};"
				echo ${sql_str_tmp}|${mysql_comm} -NB > ${tmp_file} 2>/dev/null
				if [ $? -ne 0 ]
				then
					echo -e "\033[31m[$(date "+%F %H:%M:%S")] [INFO] THREAD_${this_thread_num}:SQL语句语法错误：${sql_tmp_tmp}, 跳过该表的检查\033[0m"
					continue
				fi
				tmp_file_count="$(wc -l < ${tmp_file})"
				tmp_file_count_tmp="$(echo "${tmp_file_count}"|awk '{print int($0/2)}')"
			fi
			choose_tool_stat=0
			tail_end_count=0
			for ((k=0;k<=${tmp_file_count};k=k+${count}))
			do
				[ -f "${stop_file}" ] && exit
				steep_count="$((${k}+${count}))"
				if [ "${pri_count}x" != "0x" ]
				then
					if [ "${choose_tool_stat}x" == "0x" ]
					then
						where_str="$(head -${steep_count} ${tmp_file}|tail -${count}|tr "\n" ","|sed 's/,/),(/g'|sed 's/^/(/g;s/,($//g'|tr "\t" ","|awk -v pri_str="${pri_name}" '{print "where "pri_str" in ("$0");"}')"
					else
						where_str="$(tail -${tail_end_count} ${tmp_file}|head -${count}|tr "\n" ","|sed 's/,/),(/g'|sed 's/^/(/g;s/,($//g'|tr "\t" ","|awk -v pri_str="${pri_name}" '{print "where "pri_str" in ("$0");"}')"
						if [ ${tail_end_count} -gt ${count} ]
						then
							tail_end_count=$((${tail_end_count}-${count}))
						fi
					fi
					if [ ${steep_count} -gt ${tmp_file_count_tmp} -a ${choose_tool_stat} -eq 0 ]
					then
						choose_tool_stat=1
						tail_end_count=$((${tmp_file_count}-${steep_count}))
					fi
				else
					where_str=""
					echo -e "\033[33m[$(date "+%F %H:%M:%S")] [WARN] THREAD_${this_thread_num}:${db_name}.${table[${i}]}:没有主键。\033[0m" |tee -a ${info_log_file}
				fi
				col_str="$(${mysql_comm} -NBe "desc ${table[${i}]};" 2>/dev/null|awk '{print $1}'|tr "\n" ",")"
				col_str_isnull="$(${mysql_comm} -NBe "desc ${table[${i}]};" 2>/dev/null|awk '{print "ISNULL("$1")"}'|tr "\n" ","|sed 's#\(.*\),$#\1#g')"
				for ((j=0;j<${#host_list[@]};j++))
				do
				{
					if [ "${j}x" == "0x" ]
					then
							mysql_port="${mysql_port1}"
					else
							mysql_port="${mysql_port2}"
					fi
					mysql_comm="${mysql_path} -u${mysql_user} -p${mysql_passwd} -h${host_list[${j}]} -P${mysql_port} ${db_name} -A"
					echo "set tx_isolation='REPEATABLE-READ';SELECT '${table[${i}]}', COUNT(*) AS cnt, COALESCE(LOWER(CONV(BIT_XOR(CAST(CRC32(CONCAT_WS('#',${col_str}CONCAT(${col_str_isnull}))) AS UNSIGNED)), 10, 16)), 0) AS crc FROM ${db_name}.${table[${i}]} ${where_str};"|${mysql_comm} -NB 2>/dev/null|awk '{print $1":"$2":"$3}' > ${log_dir}/${host_list[${j}]}_${mysql_port}_${db_name}.log
				} &
				done
				wait
				if [ ${tmp_file_count} -eq 0 ]
				then
					start_count=0
				else
					[ "${k}x" == "0x" ] && start_count=1 || start_count=$((${k}+1))
				fi
				if [ $((${k}+${count}-1)) -ge ${tmp_file_count} ]
				then
					end_count="${tmp_file_count}"
				else
					end_count=$((${k}+${count}))
				fi
				if [ "$(md5sum ${log_dir}/${host_list[0]}_${mysql_port1}_${db_name}.log|awk '{print $1}')x" == "$(md5sum ${log_dir}/${host_list[1]}_${mysql_port2}_${db_name}.log|awk '{print $1}')x" ]
				then
					echo -e "\033[32m[$(date "+%F %H:%M:%S")] [OK] THREAD_${this_thread_num}:${db_name}.${table[${i}]}[${start_count},${end_count}]:数据一致。\033[0m"
				else
					echo -e "\033[31m[$(date "+%F %H:%M:%S")] [ERROR] THREAD_${this_thread_num}:${db_name}.${table[${i}]}[${start_count},${end_count}]:数据不一致。第一个节点的数据行是：$(awk -F: '{print $2}' ${log_dir}/${host_list[0]}_${mysql_port1}_${db_name}.log)，第二个节点的数据行是：$(awk -F: 'BEGIN{sum=$2}END{if (sum==""){print "0"}else {print $2}}' ${log_dir}/${host_list[1]}_${mysql_port2}_${db_name}.log)。不同部分请参考文件: ${error_log_table_dir}/${host_list[0]}_${mysql_port1}_${db_name}_${table[${i}]}.log。\033[0m" |tee -a ${error_log_file}
					echo -e "\033[33m[$(date "+%F %H:%M:%S")] [SQL] THREAD_${this_thread_num}:set tx_isolation='REPEATABLE-READ';SELECT '${table[${i}]}', COUNT(*) AS cnt, COALESCE(LOWER(CONV(BIT_XOR(CAST(CRC32(CONCAT_WS('#',${col_str}CONCAT(${col_str_isnull}))) AS UNSIGNED)), 10, 16)), 0) AS crc FROM ${db_name}.${table[${i}]} ${where_str}" >> ${error_log_table_dir}/${host_list[0]}_${mysql_port1}_${db_name}_${table[${i}]}.log
				fi
				[ -f "${log_dir}/${host_list[0]}_${mysql_port1}_${db_name}.log" ] && rm -f ${log_dir}/${host_list[0]}_${mysql_port1}_${db_name}.log
				[ -f "${log_dir}/${host_list[1]}_${mysql_port2}_${db_name}.log" ] && rm -f ${log_dir}/${host_list[1]}_${mysql_port2}_${db_name}.log
			done
			[ -f "${tmp_file}" ] && rm -f ${tmp_file}
			echo "${db_name}.${table[${i}]}" >> ${table_list_file}
		else
			echo -e "\033[33m[$(date "+%F %H:%M:%S")] [WARN] THREAD_${this_thread_num}:${db_name}.${table[${i}]}:没有主键。\033[0m" |tee -a ${error_log_file}
		fi
	done
}

f_check_diff ${db}
echo -e "\033[35m[$(date "+%F %H:%M:%S")] [INFO] THREAD_${this_thread_num}:END"
[ -f "${pid_file}" ] && rm -f ${pid_file}
[ -f "${thread_file}" ] && rm -f ${thread_file}
